package wang.conge.javasedemo.core.aio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.nio.charset.Charset;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @author conge
 * 2016年7月14日
 */
public class AsynchronousIOServer {
	private static Charset charset = Charset.forName("UTF-8");

	public static void main(String[] args) {
		int port = 10000;

		int processors = Runtime.getRuntime().availableProcessors();
		ExecutorService threadPool = Executors.newFixedThreadPool(processors);

		try {
			AsynchronousChannelGroup group = AsynchronousChannelGroup.withThreadPool(threadPool);
			AsynchronousServerSocketChannel server = AsynchronousServerSocketChannel.open(group);
			server.bind(new InetSocketAddress(port));

			doAccept(server);

			group.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
		} catch (IOException | InterruptedException e) {
			e.printStackTrace();
			System.out.println("close server");
			System.exit(0);
		}
	}

	private static void doAccept(AsynchronousServerSocketChannel server) {
		server.accept(null, new CompletionHandler<AsynchronousSocketChannel, Void>() {
			@Override
			public void completed(AsynchronousSocketChannel client, Void attachment) {
				server.accept(null, this);// accept next client connect

				doRead(client, attachment);
			}

			@Override
			public void failed(Throwable exc, Void attachment) {
				exc.printStackTrace();
			}
		});

	}

	private static void doRead(AsynchronousSocketChannel client, Void attachment) {
		ByteBuffer buffer = ByteBuffer.allocate(1024);

		client.read(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {
			@Override
			public void completed(Integer result, ByteBuffer attachment) {
				if (result <= 0) {
					try {
						System.out.println("客户端断线:" + client.getRemoteAddress().toString());
						attachment = null;
					} catch (IOException e) {
						e.printStackTrace();
					}
					return;
				}

				attachment.flip();
				String req = charset.decode(attachment).toString();
				attachment.compact();

				client.read(attachment, attachment, this);// next client read

				/** do service code **/
				System.out.println(req);

				ByteBuffer resBuffer = ByteBuffer.wrap(("response:" + req).getBytes());
				doWrite(client, resBuffer, resBuffer);
			}

			@Override
			public void failed(Throwable exc, ByteBuffer attachment) {
				exc.printStackTrace();
			}

		});
	}

	private static <V> void doWrite(AsynchronousSocketChannel client, ByteBuffer resBuffer, ByteBuffer attachment) {
		client.write(attachment, attachment, new CompletionHandler<Integer, ByteBuffer>() {

			@Override
			public void completed(Integer result, ByteBuffer attachment) {
				// TODO write success

				if (result <= 0) {
					try {
						System.out.println("客户端断线:" + client.getRemoteAddress().toString());
						attachment = null;
					} catch (IOException e) {
						e.printStackTrace();
					}
				}
			}

			@Override
			public void failed(Throwable exc, ByteBuffer attachment) {
				exc.printStackTrace();
			}
		});
	}

}
